package com.ruoyi.app.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.founder.exchange.kafka.consumer.Vistor;
import com.founder.exchange.kafka.message.ECMessage;
import com.founder.exchange.tools.ExchangeTools;
import com.ruoyi.app.domain.AppCaseInfo;
import com.ruoyi.app.domain.AppOperationLog;
import com.ruoyi.app.service.IAppCaseInfoService;
import com.ruoyi.app.service.IAppOperationLogService;
import com.ruoyi.app.service.ICaseConsumerService;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.system.domain.SysDept;
import com.ruoyi.system.service.ISysDeptService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @ Auther: chencunjun
 * @ Date: 2019/12/13 15:16
 * @ Description:
 */
@Service
public class CaseConsumerServiceImpl implements ICaseConsumerService {
    private static final Logger log = LoggerFactory.getLogger(CaseConsumerServiceImpl.class);

    @Autowired
    private IAppCaseInfoService iAppCaseInfoService;
    @Autowired
    private IAppOperationLogService iAppOperationLogService;
    @Autowired
    private ISysDeptService iSysDeptService;
    //授权码
    private final static String LICENSE_CODE_A = "S1563352928485f66e37";
    private final static String LICENSE_CODE_B = "S156335294184402dc93";
    private final static String LICENSE_CODE_C = "S156335293486031cd58";
    /**
     * kafka数据监控器(持续获取:单条)
     */
    @Override
    public void dataListener() {
        String consumerId = ExchangeTools.createConsumer(LICENSE_CODE_A);
        ExchangeTools.onMessage(consumerId, new Vistor() {
            @Override
            public void onMessage(ECMessage ecMessage) {
                 try{
                        log.info("kafka消费数据："+ecMessage.getMessageBody().toString());
                        saveORupdate(ecMessage.getMessageBody().toString());
                        ExchangeTools.commit(consumerId);//提交位移
                    }catch (Exception e){
                        log.error("同步数据异常:"+ecMessage.getMessageBody().toString()+e.getMessage());
                    }
            }
        });
    }

    /**
     * 模拟数据
     */
    @Override
    public void dataSimulator() {
        List<String> dataList = new ArrayList<>();
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401186200002017090127\",\"WSZZRQ\":\"2017\",\"WSH\":\"11111\",\"WORDOFWRIT2\":\"穗公增社戒决\",\"PZSJ\":\"2017-09-22 19:22:00.0\",\"SS_WSMC1\":\"社区戒毒决定书\",\"BA_ZZJGDM\":\"440118080300\",\"XM\":\"唐琦\",\"ZJHM\":\"430523198001034359\",\"XBDM\":\"1\",\"ZT\":\"A\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401187100002018110015\",\"WSZZRQ\":\"2018\",\"WSH\":\"14170\",\"WORDOFWRIT2\":\"穗公增强戒决\",\"PZSJ\":\"2018-11-22 10:06:00.0\",\"SS_WSMC1\":\"强制隔离戒毒决定书\",\"BA_ZZJGDM\":\"440118010000\",\"XM\":\"饶浩\",\"ZJHM\":\"433130199101090410\",\"XBDM\":\"1\",\"ZT\":\"M\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("0");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401186000002019030038\",\"WSZZRQ\":\"2019\",\"WSH\":\"2912\",\"WORDOFWRIT2\":\"穗公增强戒决\",\"PZSJ\":\"2019-03-12 10:03:00.0\",\"SS_WSMC1\":\"强制隔离戒毒决定书\",\"BA_ZZJGDM\":\"440118010000\",\"XM\":\"雍国应01\",\"ZJHM\":\"522121198702136234\",\"XBDM\":\"1\",\"ZT\":\"HR\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401186000002019030039\",\"WSZZRQ\":\"2019\",\"WSH\":\"2912\",\"WORDOFWRIT2\":\"穗公增强戒决\",\"PZSJ\":\"2019-03-12 10:03:00.0\",\"SS_WSMC1\":\"强制隔离戒毒决定书\",\"BA_ZZJGDM\":\"440118050000\",\"XM\":\"雍国应02\",\"ZJHM\":\"522121198702136235\",\"XBDM\":\"1\",\"ZT\":\"HR\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401186000002019030011\",\"WSZZRQ\":\"2019\",\"WSH\":\"2912\",\"WORDOFWRIT2\":\"穗公增强戒决\",\"PZSJ\":\"2019-03-12 10:03:00.0\",\"SS_WSMC1\":\"强制隔离戒毒决定书\",\"BA_ZZJGDM\":\"440118010100\",\"XM\":\"雍国应03\",\"ZJHM\":\"522121198702136235\",\"XBDM\":\"1\",\"ZT\":\"HR\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401186000002019100204\",\"WSZZRQ\":\"2019\",\"WSH\":\"12325\",\"WORDOFWRIT2\":\"穗公增社戒决\",\"PZSJ\":\"2019-11-01 10:55:00.0\",\"SS_WSMC1\":\"社区戒毒决定书\",\"BA_ZZJGDM\":\"440118600000\",\"XM\":\"赵明理\",\"ZJHM\":\"430424198305012336\",\"XBDM\":\"1\",\"ZT\":\"HL\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401186000002019100242\",\"WSZZRQ\":\"2019\",\"WSH\":\"12544\",\"WORDOFWRIT2\":\"穗公增社戒决\",\"PZSJ\":\"2019-11-08 19:24:00.0\",\"SS_WSMC1\":\"社区戒毒决定书\",\"BA_ZZJGDM\":\"440118600000\",\"XM\":\"刘敏\",\"ZJHM\":\"421023198607114910\",\"XBDM\":\"1\",\"ZT\":\"A\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401185500002019010061\",\"WSZZRQ\":\"2019\",\"WSH\":\"1476\",\"WORDOFWRIT2\":\"穗公增强戒决\",\"PZSJ\":\"2019-01-26 17:08:00.0\",\"SS_WSMC1\":\"强制隔离戒毒决定书\",\"BA_ZZJGDM\":\"440118550000\",\"XM\":\"王光耀\",\"ZJHM\":\"441324199212090337\",\"XBDM\":\"1\",\"ZT\":\"A\",\"FZJG_ZZJGDM\":\"粤A\"}]}");
        dataList.add("{\"chname\":[\"案件编号\",\"文书年份\",\"文书号\",\"文书字\",\"批准时间\",\"诉讼文书名称1\",\"办案单位\",\"姓名(姓名)\",\"身份证号码(证件号码)\",\"性别(性别代码)\",\"驾驶证状态\",\"发证机关(发证机关_组织机构代码)\"],\"data\":[{\"AJBH\":\"A4401187100002019040010\",\"WSZZRQ\":\"2019\",\"WSH\":\"4146\",\"WORDOFWRIT2\":\"穗公增强戒决\",\"PZSJ\":\"2019-04-03 23:57:00.0\",\"SS_WSMC1\":\"强制隔离戒毒决定书\",\"BA_ZZJGDM\":\"440118710000\",\"XM\":\"钟分爱\",\"ZJHM\":\"441322198008174923\",\"XBDM\":\"2\",\"ZT\":\"A\",\"FZJG_ZZJGDM\":\"粤A\"}]}");

        /**
         * 模拟循环获取数据
         */
       while(true){
           for(String data:dataList){
               saveORupdate(data);
           }
           try {
               Thread.sleep(10000);  //线程休眠10s
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
    }

    /**
     * 单次接收数据(批量获取)
     */
    @Override
    public void dataRecevie() {
        //消费者id
        String consumerId_A = ExchangeTools.createConsumer(LICENSE_CODE_A);
        String consumerId_B = ExchangeTools.createConsumer(LICENSE_CODE_B);
        String consumerId_C = ExchangeTools.createConsumer(LICENSE_CODE_C);
        while (true) {
            List<ECMessage> ecMessage_A = ExchangeTools.recevie(consumerId_A);
            log.info("kafka消费组："+LICENSE_CODE_A+"，消费数据长度："+ecMessage_A.size());
            List<ECMessage> ecMessage_B= ExchangeTools.recevie(consumerId_B);
            log.info("kafka消费组："+LICENSE_CODE_B+"，消费数据长度："+ecMessage_B.size());
            List<ECMessage> ecMessage_C = ExchangeTools.recevie(consumerId_C);
            log.info("kafka消费组："+LICENSE_CODE_C+"，消费数据长度："+ecMessage_C.size());
            List<ECMessage> results=new ArrayList<>();
            results.addAll(ecMessage_A);
            results.addAll(ecMessage_B);
            results.addAll(ecMessage_C);
            for (ECMessage message : results) {
                //打印获取的数据
                String msg = message.getMessageBody().toString();
                log.info("kafka消费数据："+msg);
                saveORupdate(msg);
            }
            //提交kafka位移
            if(!ecMessage_A.isEmpty()){
                ExchangeTools.commit(consumerId_A);
            }
            if(!ecMessage_B.isEmpty()){
                ExchangeTools.commit(consumerId_B);
            }
            if(!ecMessage_C.isEmpty()){
                ExchangeTools.commit(consumerId_C);
            }
            try {
                Thread.sleep(300000);  //线程休眠60s
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void saveORupdate(String data) {

        try{
            JSONObject jsonObject = JSON.parseObject(data);
            JSONArray jsonArray = JSONArray.parseArray(jsonObject.get("data").toString());
            JSONObject vo = JSON.parseObject(jsonArray.get(0).toString());

            //AppCaseInfo searchCase = iAppCaseInfoService.selectAppCaseInfoByCaseCode(vo.getString("AJBH"));
            AppCaseInfo searchCase = iAppCaseInfoService.selectAppCaseInfoBySearch(vo.getString("AJBH"),vo.getString("WSH"),vo.getString("SS_WSMC1"));
            String caseCompanyName = "外地公安机关";

            if (null != searchCase) {
                searchCase.setState("0");
                searchCase.setCreateTime(new Date());
                iAppCaseInfoService.updateAppCaseInfo(searchCase);
            }else{
                AppCaseInfo caseInfo = new AppCaseInfo();
                if(null!=vo.getString("BA_ZZJGDM")){
                    SysDept search = new SysDept();
                    search.setDeptCode(vo.getString("BA_ZZJGDM"));
                    List<SysDept> sysDepts = iSysDeptService.selectAppDeptList(search);
                    if(StringUtils.isNotEmpty(sysDepts)){
                        caseCompanyName = sysDepts.get(0).getDeptName();
                    }
                    caseInfo.setCaseCompanyCode(vo.getString("BA_ZZJGDM")); //办案单位编码
                    caseInfo.setCaseCompanyName(caseCompanyName); //办案单位编码
                }
                caseInfo.setCaseCompanyCode(vo.getString("BA_ZZJGDM")); //办案单位编码
                caseInfo.setCaseCode(vo.getString("AJBH")); //案件号
                caseInfo.setClericalYear(vo.getString("WSZZRQ")); //年书年份
                caseInfo.setClericalCode(vo.getString("WSH"));  //文书号
                caseInfo.setClericalWord(vo.getString("WORDOFWRIT2")); //文书字
                caseInfo.setApprovalTime(vo.getString("PZSJ")); //文书批准时间
                caseInfo.setLitiClericalWord(vo.getString("SS_WSMC1")); //诉讼文书名称

                caseInfo.setPeopleName(vo.getString("XM")); //执行人姓名
                caseInfo.setPeopleCard(vo.getString("ZJHM")); //执行人身份证
                caseInfo.setPeopleSex(vo.getString("XBDM")); //执行人性别
                caseInfo.setDriveLicenseState(vo.getString("ZT")); //驾驶证状态
                caseInfo.setCertifiAuthority(vo.getString("FZJG_ZZJGDM")); //发证机关
                caseInfo.setCreateTime(new Date());  //创建时间
                caseInfo.setState("0"); //状态
                iAppCaseInfoService.insertAppCaseInfo(caseInfo);
                saveLog(caseInfo.getId(),caseInfo.getPeopleName());
            }
        } catch (JSONException e){
            e.printStackTrace();
        }


    }

    private void saveLog(Long caseId,String peopleName){
        AppOperationLog appOperationLog = new AppOperationLog();
        appOperationLog.setOperaType("addCaseInfo");
        appOperationLog.setOperaContent("新增未注销人员"+peopleName);
        appOperationLog.setOperaTime(new Date());
        appOperationLog.setSourceId(caseId);
        appOperationLog.setState(0);
        appOperationLog.setUserId(0L);
        appOperationLog.setUserName("系统同步");
        iAppOperationLogService.insertAppOperationLog(appOperationLog);
    }
}
